package xuecheng.tool.mq.redis;

import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.springframework.scheduling.annotation.Async;

import lombok.extern.slf4j.Slf4j;
import xuecheng.tool.core.datetime.DateTimeUtils;
import xuecheng.tool.json.JsonUtils;
import xuecheng.tool.json.JsonUtilsFactory;
import xuecheng.tool.mq.MqListener;
import xuecheng.tool.nosql.redis.RedisUtils;

/**
 * 订阅任务
 *
 * @author PengXueCheng
 * @version 1.0.0
 */
@Slf4j
class ConsumerFactory {

    private final JsonUtils jsonUtils = JsonUtilsFactory.obtain();

    private RedisUtils redisUtils;
    private RedisMqUtils redisMqUtils;

    ConsumerFactory(RedisUtils redisUtils, RedisMqUtils redisMqUtils) {
        this.redisUtils = redisUtils;
        this.redisMqUtils = redisMqUtils;
    }

    @Async
    public <T> void start(String topic, Class<T> messageClass, MqListener<T> listener) {
        // 延迟1分钟后再启动
        try {
            Thread.sleep(60 * 1000);
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        while (true) {
            long t1 = System.nanoTime();
            String messageJson = redisUtils.lBRightPop(topic, 1, TimeUnit.SECONDS);
            long t2 = System.nanoTime();
            if (StringUtils.isEmpty(messageJson)) {
                continue;
            }
            // 控制台显示的消息
            String consoleShowMsg = messageJson;
            // 控制台消息显示的最大长度
            int maxConsoleShowMsgLen = 100;
            if (messageJson.length() > maxConsoleShowMsgLen) {
                consoleShowMsg = String.format("%s...", messageJson.substring(0, maxConsoleShowMsgLen));
            }
            log.info("接收到消息. 耗时: {}ms, topic = {}, message = {}", DateTimeUtils.getConsumeTime(t1, t2), topic,
                consoleShowMsg);
            T t;
            try {
                t = jsonUtils.fromJsonAsObject(messageJson, messageClass);
            } catch (Exception e) {
                log.error(messageJson + "_" + messageClass + "_" + e.getMessage(), e);
                // 消息消费失败, 需要重新投递, 继续消费其他消息
                redisMqUtils.send(topic, messageJson);
                continue;
            }
            try {
                listener.onMessage(t);
            } catch (Exception e) {
                log.error(listener.toString() + "_" + t.toString() + "_" + e.getMessage(), e);
                // 消息消费失败, 需要重新投递, 继续消费其他消息
                redisMqUtils.send(topic, messageJson);
            }
        }
    }

}
